// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: authentication-free shared key and token derivation between clients
// feature: change E2E packet format
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: interval report metrics
// feature: support log Event and realtime metrics
// feature: E2E Profile04 and Configuration Function Development
// ------------------------------------------------------------------

#include <history/CacheChangePool.h>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <edds/rtps/writer/RTPSWriter.h>
#include <edds/dds/core/policy/ParameterSerializer.hpp>

#include <history/WriterHistory.h>

#include <edds/rtps/messages/RTPSMessageCreator.h>

#include <mutex>

#include <elog/Log.hpp>
#include <history/BasicPayloadPool.hpp>

#include "../flowcontrol/FlowController.hpp"

using namespace vbs::common;
namespace evbs {
namespace ertps {
namespace rtps {

RTPSWriter::RTPSWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                       edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen)
    : Endpoint(impl, guid, att.endpoint),
      flow_controller_(flow_controller),
      mp_history(hist),
      mp_listener(listen),
      is_async_(att.mode == SYNCHRONOUS_WRITER ? false : true),
      liveliness_kind_(att.liveliness_kind),
      liveliness_lease_duration_(att.liveliness_lease_duration),
      liveliness_announcement_period_(att.liveliness_announcement_period),
      enable_e2e_protection_(att.enable_e2e_protection),
      e2e_p04_min_data_length_(att.e2e_p04_min_data_length),
      e2e_p04_max_data_length_(att.e2e_p04_max_data_length),
      e2e_p04_max_delta_counter_(att.e2e_p04_max_delta_counter),
      trace_point_status_(att.trace_point_status) {
    PoolConfig cfg = PoolConfig::from_history_attributes(hist->m_att);
    std::shared_ptr<IChangePool> change_pool;
    std::shared_ptr<IPayloadPool> payload_pool;
    payload_pool = BasicPayloadPool::get(cfg, change_pool);

    is_send_multi = att.is_send_multi;
    topic_with_key_ = (att.endpoint.topicKind == TopicKind_t::WITH_KEY);

    init(payload_pool, change_pool, att);
}

RTPSWriter::RTPSWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                       const std::shared_ptr<IPayloadPool>& payload_pool, edds::rtps::FlowController* flow_controller,
                       WriterHistory* hist, WriterListener* listen)
    : RTPSWriter(impl, guid, att, payload_pool,
                 std::make_shared<CacheChangePool>(PoolConfig::from_history_attributes(hist->m_att)), flow_controller,
                 hist, listen) {}

RTPSWriter::RTPSWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
                       const std::shared_ptr<IPayloadPool>& payload_pool,
                       const std::shared_ptr<IChangePool>& change_pool, edds::rtps::FlowController* flow_controller,
                       WriterHistory* hist, WriterListener* listen)
    : Endpoint(impl, guid, att.endpoint),
      flow_controller_(flow_controller),
      mp_history(hist),
      mp_listener(listen),
      is_async_(att.mode == SYNCHRONOUS_WRITER ? false : true),
      liveliness_kind_(att.liveliness_kind),
      liveliness_lease_duration_(att.liveliness_lease_duration),
      liveliness_announcement_period_(att.liveliness_announcement_period),
      enable_e2e_protection_(att.enable_e2e_protection),
      e2e_p04_min_data_length_(att.e2e_p04_min_data_length),
      e2e_p04_max_data_length_(att.e2e_p04_max_data_length),
      e2e_p04_max_delta_counter_(att.e2e_p04_max_delta_counter),
      trace_point_status_(att.trace_point_status) {
    init(payload_pool, change_pool, att);
}

void RTPSWriter::init(const std::shared_ptr<IPayloadPool>& payload_pool,
                      const std::shared_ptr<IChangePool>& change_pool, const WriterAttributes& att) {
    UNUSED_PARAMETER(att);
    payload_pool_ = payload_pool;
    change_pool_ = change_pool;
    fixed_payload_size_ = 0U;
    if (mp_history->m_att.memoryPolicy == PREALLOCATED_MEMORY_MODE) {
        fixed_payload_size_ = mp_history->m_att.payloadMaxSize;
    }

    //mp_history->mp_writer = this;
    //mp_history->mp_mutex  = &mp_mutex;
    //mp_mutex = mp_history->getMutex();
    mp_history->guid = m_guid;
    mp_history->topicName = getTopicName();
#if 0
    mp_history->notify_writer = std::bind(&RTPSWriter::unsent_change_added_to_history, this,
                                                   std::placeholders::_1, std::placeholders::_2);
#endif
    mp_history->change_removed_by_history =
        std::bind(&RTPSWriter::change_removed_by_history, this, std::placeholders::_1);
    mp_history->getMaxDataSize = [this](CacheChange_t* change) {
        return getMaxDataSize(change);
    };
    mp_history->payload_pool_ = payload_pool;
    mp_history->change_pool_ = change_pool;
    flow_controller_->register_writer(this);
    logDebug(RTPS_WRITER, "RTPSWriter created");
}

RTPSWriter::~RTPSWriter() {
    logDebug(RTPS_WRITER, "RTPSWriter destructor"
                              << " topic " << getTopicName());

    // Deletion of the events has to be made in child destructor.
    // Also at this point all CacheChange_t must have been released by the child destructor

    //mp_history->mp_writer = nullptr;
    //mp_history->mp_mutex  = nullptr;
}

void RTPSWriter::deinit() {
    // First, unregister changes from FlowController. This action must be protected.
    std::vector<CacheChange_t*> changes_to_release;
    {
        std::lock_guard<RecursiveTimedMutex> guard(mp_history->getMutex());
        for (auto it = mp_history->changesBegin(); it != mp_history->changesEnd();) {
            CacheChange_t* change = *it;
            it = mp_history->remove_change(it, false);
            changes_to_release.push_back(change);
        }
    }

    for (auto change : changes_to_release) {
        flow_controller_->remove_change(change);
    }

    flow_controller_->unregister_writer(this);

    for (auto change : changes_to_release) {
        mp_history->safe_release_cache(change);
    }
}

bool RTPSWriter::set_listener(WriterListener* const listener) {
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    mp_listener = listener;
    return true;
}

CacheChange_t* RTPSWriter::new_change(const std::function<uint32_t()>& dataCdrSerializedSize, ChangeKind_t changeKind,
                                      InstanceHandle_t handle) {
    std::unique_lock<RecursiveTimedMutex> guard(mp_history->getMutex());
    CacheChange_t* reserved_change = nullptr;
    if (!change_pool_->reserve_cache(reserved_change)) {
        logWarning(RTPS_WRITER, "Topic " << getTopicName() << " Problem reserving cache from pool");
        guard.unlock();
        return nullptr;
    }

    uint32_t payload_size = fixed_payload_size_ ? fixed_payload_size_ : dataCdrSerializedSize();
    if (!payload_pool_->get_payload(payload_size, *reserved_change)) {
        (void)change_pool_->release_cache(reserved_change);
        logWarning(RTPS_WRITER, "Topic " << getTopicName() << " Problem reserving payload from pool");
        guard.unlock();
        return nullptr;
    }
    guard.unlock();
    reserved_change->kind = changeKind;
    if ((m_att.topicKind == WITH_KEY) && (!handle.isDefined())) {
        logWarning(RTPS_WRITER, "Changes in KEYED Writers need a valid instanceHandle");
    }
    reserved_change->instanceHandle = handle;
    reserved_change->writerGUID = m_guid;
    reserved_change->writer_info.previous = nullptr;
    reserved_change->writer_info.next = nullptr;
    reserved_change->writer_info.num_sent_submessages = 0U;
    return reserved_change;
}

CacheChange_t* RTPSWriter::new_change(ChangeKind_t changeKind, InstanceHandle_t handle) {
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    CacheChange_t* reserved_change = nullptr;
    if (!change_pool_->reserve_cache(reserved_change)) {  // NOLINT
        logWarning(RTPS_WRITER, "Topic " << getTopicName() << " Problem reserving cache from pool");
        return nullptr;
    }
    reserved_change->kind = changeKind;
    if (m_att.topicKind == WITH_KEY && !handle.isDefined()) {
        logWarning(RTPS_WRITER, "Topic " << getTopicName() << " Changes in KEYED Writers need a valid instanceHandle");
    }
    reserved_change->instanceHandle = handle;
    reserved_change->writerGUID = m_guid;
    reserved_change->writer_info.previous = nullptr;
    reserved_change->writer_info.next = nullptr;
    reserved_change->writer_info.num_sent_submessages = 0U;
    return reserved_change;
}

bool RTPSWriter::release_change(CacheChange_t* change) {
    // Asserting preconditions
    assert(change != nullptr);
    assert(change->writerGUID == m_guid);

    std::lock_guard<RecursiveTimedMutex> guard(mp_history->getMutex());

    IPayloadPool* pool = change->payload_owner();
    if (pool) {
        (void)pool->release_payload(*change);
    }
    return change_pool_->release_cache(change);
}

SequenceNumber_t RTPSWriter::get_seq_num_min() {
    CacheChange_t* change;
    if (mp_history->get_min_change(&change) && (change != nullptr)) {
        return change->sequenceNumber;
    } else {
        return c_SequenceNumber_Unknown;
    }
}

SequenceNumber_t RTPSWriter::get_seq_num_max() {
    return mp_history->get_max_change();
}

#ifdef BATCH_SEND_ENABLE
SequenceNumber_t RTPSWriter::get_last_seq_from_min_batch() {
    CacheChange_t* change;
    if (mp_history->get_last_change_from_min_batch(&change) && (change != nullptr)) {
        return change->sequenceNumber;
    } else {
        return c_SequenceNumber_Unknown;
    }
}
#endif

uint32_t RTPSWriter::getTypeMaxSerialized() {
    return mp_history->getTypeMaxSerialized();
}

bool RTPSWriter::remove_older_changes(uint32_t max) {
    logDebug(RTPS_WRITER, "Starting process clean_history for writer " << getGuid());
    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
    bool limit = (max != 0U);

    bool remove_ret = mp_history->remove_min_change();
    bool at_least_one = remove_ret;
    uint32_t count = 1U;

    while (remove_ret && ((!limit) || (count < max))) {
        remove_ret = mp_history->remove_min_change();
        ++count;
    }

    return at_least_one;
}

constexpr uint32_t info_secure_token_message_length = 8U;
constexpr uint32_t info_dst_message_length = 16U;
constexpr uint32_t info_ts_message_length = 12U;
constexpr uint32_t data_frag_submessage_header_length = 36U;
constexpr uint32_t heartbeat_message_length = 32U;

uint32_t RTPSWriter::getInlineQosSize(const CacheChange_t* change) {
    // Calc additional size for inline QoS
    uint32_t inline_qos_size = change->inline_qos.length;
    if (change->write_params.related_sample_identity() != SampleIdentity::unknown()) {
        inline_qos_size += (2 * edds::dds::ParameterSerializer<Parameter_t>::PARAMETER_SAMPLE_IDENTITY_SIZE);
    }

    if (ChangeKind_t::ALIVE != change->kind && topic_with_key_) {
        inline_qos_size += edds::dds::ParameterSerializer<Parameter_t>::PARAMETER_KEY_SIZE;
        inline_qos_size += edds::dds::ParameterSerializer<Parameter_t>::PARAMETER_STATUS_SIZE;
    }

    // If inlineqos for related_sample_identity is required, then remove its size from the final fragment size.
    if (0 < inline_qos_size) {
        inline_qos_size += edds::dds::ParameterSerializer<Parameter_t>::PARAMETER_SENTINEL_SIZE;
    }
    return inline_qos_size;
}

uint32_t RTPSWriter::getMaxDataSize(const CacheChange_t* change) {
    uint32_t flow_max = flow_controller_->get_max_payload();
    uint32_t part_max = mp_RTPSParticipant->getMaxMessageSize();
    uint32_t max_size = flow_max > part_max ? part_max : flow_max;

    max_size = calculateMaxDataSize(max_size);
    uint32_t inline_qos_size = getInlineQosSize(change);
    if (max_size < inline_qos_size) {
        return 0;
    } else {
        max_size -= inline_qos_size;
    }
    return max_size &= ~3U;
}

uint32_t RTPSWriter::calculateMaxDataSize(uint32_t length) {
    uint32_t maxDataSize = mp_RTPSParticipant->calculateMaxDataSize(length);

    maxDataSize -= info_dst_message_length + info_ts_message_length + data_frag_submessage_header_length +
                   heartbeat_message_length;

    // TODO(Ricardo) inlineqos in future.

#if HAVE_SECURITY
    if (mp_RTPSParticipant->security_manager().authTokenAlwaysOn || mp_RTPSParticipant->is_secure()) {
        maxDataSize -= info_secure_token_message_length;
    }
    if (getAttributes().security_attributes().is_submessage_protected) {
        maxDataSize -= mp_RTPSParticipant->security_manager().calculate_extra_size_for_rtps_submessage(m_guid);
    }

    if (getAttributes().security_attributes().is_payload_protected) {
        maxDataSize -= mp_RTPSParticipant->security_manager().calculate_extra_size_for_encoded_payload(m_guid);
    }
#endif  // if HAVE_SECURITY
    return maxDataSize;
}

void RTPSWriter::add_guid(LocatorSelectorSender& locator_selector, const GUID_t& remote_guid) {
    const GuidPrefix_t& prefix = remote_guid.guidPrefix;
    (void)locator_selector.all_remote_readers.push_back(remote_guid);
    if (std::find(locator_selector.all_remote_participants.begin(), locator_selector.all_remote_participants.end(),
                  prefix) == locator_selector.all_remote_participants.end()) {
        (void)locator_selector.all_remote_participants.push_back(prefix);
    }
}

void RTPSWriter::compute_selected_guids(LocatorSelectorSender& locator_selector) {
    locator_selector.all_remote_readers.clear();
    locator_selector.all_remote_participants.clear();

    for (LocatorSelectorEntry* entry : locator_selector.locator_selector.transport_starts()) {
        if (entry->enabled) {
            add_guid(locator_selector, entry->remote_guid);
        }
    }
}

void RTPSWriter::update_cached_info_nts(LocatorSelectorSender& locator_selector) {
    locator_selector.locator_selector.reset(true);
    mp_RTPSParticipant->select_locators(locator_selector.locator_selector);
}

const LivelinessQosPolicyKind& RTPSWriter::get_liveliness_kind() const {
    return liveliness_kind_;
}

const Duration_t& RTPSWriter::get_liveliness_lease_duration() const {
    return liveliness_lease_duration_;
}

const Duration_t& RTPSWriter::get_liveliness_announcement_period() const {
    return liveliness_announcement_period_;
}

bool RTPSWriter::is_pool_initialized() const {
    return true;
}

bool RTPSWriter::send_nts(CDRMessage_t* message, const LocatorSelectorSender& locator_selector) const {
    RTPSParticipantImpl* participant = getRTPSParticipant();

    return (locator_selector.locator_selector.selected_size() == 0U) ||
           participant->sendSync(message, m_guid, locator_selector.locator_selector.begin(),
                                 locator_selector.locator_selector.end(), is_send_multi);
}

bool RTPSWriter::change_is_acked_or_fully_delivered(const CacheChange_t* change) {
    bool is_acked = false;
    if (get_disable_positive_acks()) {
        is_acked = has_been_fully_delivered(change->sequenceNumber);
    } else {
        is_acked = is_acked_by_all(change);
    }
    return is_acked;
}

void RTPSWriter::add_announce_locator(Locator_t loc) {
    EndpointAttributes& attr = getAttributes();
    logDebug(RTPS_READER, "Writer add announce locator: " << loc);
    if (loc.kind == LOCATOR_PORT_INVALID) {
        logDebug(RTPS_READER, "locator is invalid");
        LocatorList_t locators = getRTPSParticipant()->getRTPSParticipantAttributes().defaultUnicastLocatorList;
        for (auto loc : locators) {
            attr.unicastLocatorList.push_back(loc);
        }
        return;
    }
    attr.unicastLocatorList.push_back(loc);
}

}  // namespace rtps
}  // namespace ertps

}  // namespace evbs
